package com.example.demo.rabbitMq;

import lombok.AllArgsConstructor;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMq配置类
 */
@Configuration
@AllArgsConstructor
public class RabbitMqConfig {

    private AmqpAdmin amqpAdmin;

    @Bean
    public void init() {
        /** 直连交换机 */
        directExchangeInit(RabbitMqConstants.DirectQueue, RabbitMqConstants.DirectExchange, RabbitMqConstants.DirectRoutingKey);
        /** 死信队列初始化 */
        topicExchangeInit(RabbitMqConstants.SX_Queue, RabbitMqConstants.SX_Exchange_Value, RabbitMqConstants.SX_Routing_Value, false, "", "");
        /** 主体交换机 */
        topicExchangeInit(RabbitMqConstants.TopicQueue, RabbitMqConstants.TopicExchange, RabbitMqConstants.TopicRoutingKey,
                true, RabbitMqConstants.SX_Exchange_Value, RabbitMqConstants.SX_Routing_Value);
        topicExchangeInit(RabbitMqConstants.TopicQueueFirst, RabbitMqConstants.TopicExchange, RabbitMqConstants.TopicRoutingKeyFirst, false, "", "");
        topicExchangeInit(RabbitMqConstants.TopicQueueSecond, RabbitMqConstants.TopicExchange, RabbitMqConstants.TopicRoutingKeySecond, false, "", "");

        /** 延迟消息 */
        delayedInit(RabbitMqConstants.Delayed_Queue, RabbitMqConstants.Delay_Exchange, RabbitMqConstants.Delay_Routing);
    }

    /** 配置发送确认回调 */
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        /** 设置连接工厂 */
        rabbitTemplate.setConnectionFactory(connectionFactory);
        /** 设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函 */
        rabbitTemplate.setMandatory(true);

        /**
         * 推送消息存在四种情况
         * 1消息推送到server，但是在server里找不到交换机      只触发ConfirmCallback
         * 2消息推送到server，找到交换机了，但是没找到队列     触发ConfirmCallback和ReturnCallback
         * 3消息推送到sever，交换机和队列啥都没找到           只触发ConfirmCallback
         * 4消息推送成功                                     只触发ConfirmCallback
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            System.out.println("ConfirmCallback:     "+"相关数据：" + correlationData);
            System.out.println("ConfirmCallback:     "+"确认情况：" + ack);
            System.out.println("ConfirmCallback:     "+"原因：    " + cause);
        });

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            System.out.println("ReturnCallback:     "+"消息：    "+message);
            System.out.println("ReturnCallback:     "+"回应码：  "+replyCode);
            System.out.println("ReturnCallback:     "+"回应信息："+replyText);
            System.out.println("ReturnCallback:     "+"交换机：  "+exchange);
            System.out.println("ReturnCallback:     "+"路由键：  "+routingKey);
        });

        return rabbitTemplate;
    }

    /** 定义直连交换机通用绑定方法 */
    private void directExchangeInit(String queueName, String exchangeName, String routingKey) {
        /** 创建队列 */
        Queue queue = new Queue(queueName);
        /** 创建直连交换机 */
        DirectExchange directExchange = new DirectExchange(exchangeName);
        /** 绑定交换机与队列 */
        Binding binding = BindingBuilder.bind(queue).to(directExchange).with(routingKey);

        amqpAdmin.declareQueue(queue);
        amqpAdmin.declareExchange(directExchange);
        amqpAdmin.declareBinding(binding);
    }

    /** 定义主体交换机通用绑定方法 */
    private void topicExchangeInit(String queueName, String exchangeName, String routingKey, boolean hasSX, String sxExchange, String sxRouting) {
        /** 创建队列 */
        Queue queue;
        if (hasSX) {
            /** 设置死信交换机及路由键 */
            Map<String, Object> args = new HashMap<>();
            args.put(RabbitMqConstants.SX_Exchange_Key, sxExchange);
            args.put(RabbitMqConstants.SX_Routing_Key, sxRouting);
            /** x-message-ttl可以设置队列消息过期时间，不灵活，不使用 */
            queue = QueueBuilder.durable(queueName).withArguments(args).build();
        } else {
            queue = new Queue(queueName);
        }
        /** 创建直连交换机 */
        TopicExchange topicExchange = new TopicExchange(exchangeName);
        /** 绑定交换机与队列 */
        Binding binding = BindingBuilder.bind(queue).to(topicExchange).with(routingKey);

        amqpAdmin.declareQueue(queue);
        amqpAdmin.declareExchange(topicExchange);
        amqpAdmin.declareBinding(binding);
    }

    /** 初始化延迟消息队列 */
    private void delayedInit(String queueName, String exchangeName, String routingKey) {
        Map<String, Object> args = new HashMap<>();
        args.put(RabbitMqConstants.Delayed_Type, RabbitMqConstants.Delayed_Type_Value);
        /** 这里要特别注意的是，使用的是CustomExchange,不是DirectExchange，另外CustomExchange的类型必须是x-delayed-message */
        CustomExchange customExchange = new CustomExchange(exchangeName, RabbitMqConstants.Delayed_Message, true, false, args);
        Queue queue = new Queue(queueName);
        Binding binding = BindingBuilder.bind(queue).to(customExchange).with(routingKey).noargs();

        amqpAdmin.declareQueue(queue);
        amqpAdmin.declareExchange(customExchange);
        amqpAdmin.declareBinding(binding);
    }

}
